package com.streamsets.datacollector.pileline.stage;

import com.streamsets.datacollector.pileline.enums.HiveHadoopFSConfigEnum;
import com.streamsets.datacollector.pileline.stage.dto.HadoopFSConfigDto;
import com.streamsets.datacollector.restapi.bean.ConfigConfigurationJson;
import com.streamsets.datacollector.restapi.bean.StageConfigurationJson;

import java.util.ArrayList;
import java.util.List;

/**
 * 初始化HadoopFS配置
 * @author wengzhonghui
 * @date 2019/10/16 19:12
 */
public class InitHadoopFSConfig {

    /*
     * stage instanceName名称
     */
    public static final String STAGE_NAME = "HadoopFS_01";

    public static void init(StageConfigurationJson stage, HadoopFSConfigDto hdfsDto){
        if(STAGE_NAME.equalsIgnoreCase(stage.getInstanceName())) {
            if (hdfsDto != null) {
                List<ConfigConfigurationJson> configs = intConfig(stage.getConfiguration(), hdfsDto);
                stage.setConfiguration(configs);
            }
        }
    }

    public static List<ConfigConfigurationJson> intConfig(List<ConfigConfigurationJson> oldConfigs, HadoopFSConfigDto hdfsDto){
        List<ConfigConfigurationJson> newConfigs = new ArrayList<>();
        for(ConfigConfigurationJson config :oldConfigs ){
            config = initHdfsUri(config, hdfsDto.getHdfsUri());
            config = initMaxRecordsPerFile(config, hdfsDto.getMaxRecordsPerFile());
            config = initMaxFileSize(config, hdfsDto.getMaxFileSize());
            config = initIdleTimeout(config, hdfsDto.getIdleTimeout());
            config = initLateRecordsLimit(config, hdfsDto.getLateRecordsLimit());
            config = initDataFormat(config, hdfsDto.getDataFormat());
            newConfigs.add(config);
        }
        return newConfigs;
    }


    // 初始化hdfsUri
    private static ConfigConfigurationJson initHdfsUri(ConfigConfigurationJson config, String hdfsUri){
        if(HiveHadoopFSConfigEnum.hdfsUri.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(HiveHadoopFSConfigEnum.hdfsUri.getName()
                    ,hdfsUri);
        }
        return config;
    }

    private static ConfigConfigurationJson initMaxRecordsPerFile(ConfigConfigurationJson config, Integer maxRecordsPerFile){
        if(HiveHadoopFSConfigEnum.maxRecordsPerFile.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(HiveHadoopFSConfigEnum.maxRecordsPerFile.getName()
                    ,maxRecordsPerFile);
        }
        return config;
    }

    private static ConfigConfigurationJson initMaxFileSize(ConfigConfigurationJson config, Integer maxFileSize){
        if(HiveHadoopFSConfigEnum.maxFileSize.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(HiveHadoopFSConfigEnum.maxFileSize.getName()
                    ,maxFileSize);
        }
        return config;
    }


    private static ConfigConfigurationJson initIdleTimeout(ConfigConfigurationJson config, String idleTimeout){
        idleTimeout = "${"+ idleTimeout +"* SECONDS}";
        if(HiveHadoopFSConfigEnum.idleTimeout.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(HiveHadoopFSConfigEnum.idleTimeout.getName()
                    ,idleTimeout);
        }
        return config;
    }

    private static ConfigConfigurationJson initLateRecordsLimit(ConfigConfigurationJson config, String lateRecordsLimit){
        lateRecordsLimit = "${"+ lateRecordsLimit +"* SECONDS}";
        if(HiveHadoopFSConfigEnum.lateRecordsLimit.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(HiveHadoopFSConfigEnum.lateRecordsLimit.getName()
                    ,lateRecordsLimit);
        }
        return config;
    }

    private static ConfigConfigurationJson initDataFormat(ConfigConfigurationJson config, String dataFormat){
        if(HiveHadoopFSConfigEnum.dataFormat.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(HiveHadoopFSConfigEnum.dataFormat.getName()
                    ,dataFormat);
        }
        return config;
    }

}
